const crypto = require("crypto");
const ConnectionPool = require("jackpot");
const net = require("net");
const {TaskRunner} = require("concurrent-tasks");
// 新建并发任务
const runner = new TaskRunner();
// task map
const taskMap = {};

const SocketNioClient = function() {
  this.headByte = Buffer.of(0xa5);
  this.secretByte = Buffer.of(0xff);
  this.secretIv = new Buffer.from("");
  this.secretAlgorithm = "aes-128-ecb";
  this.heartbeatBytes = new Buffer.from(
    JSON.stringify({
      code: 200,
      method: "",
      nodeData: {
        node: null,
      },
    })
  );
  this.ackPromiseCallbackMap = {};
  this.syncPromiseCallbackMap = {};
  this.methodMap = {};
  return this;
};

SocketNioClient.prototype.build = function(options) {
  const that = this;
  this.port = options.port;
  this.host = options.host;
  this.appKeyBytes = Buffer.from(options.appKey);
  this.secretBytes = Buffer.from(options.secret);
  this.sizeLimit = 100 * 1024 * 1024;
  this.heartbeatInterval = (options.heartbeatInterval || 25) * 1000;
  this.connection = options.connection
    ? options.connection
    : function(conn) {
        console.log("连接成功");
      };
  this.pool = new ConnectionPool(
    2,
    function() {
      console.log('创建连接池')
      let socket = new net.Socket();
      socket.setKeepAlive(true);
      socket.connect(that.port, that.host, function() {
        that.connection(socket);
      });
      socket.on("data", function(data) {
        that.read(data, this);
      });
      socket.on("error", function(error) {
        console.log("连接异常:", error.message);
      });
      socket.on("end", function() {
        console.log("连接关闭");
      });
      return socket;
    },
    {
      retries: 3,
      max: "2 seconds",
      min: "2 seconds",
    }
  );
  this.intervalId = setInterval(() => {
    try {
      console.log('setInterval')
      that.writeBytes(that.packageFunc(that.heartbeatBytes, false));
    } catch (e) {
      console.warn("客户端心跳异常", e);
    }
  }, this.heartbeatInterval);
  return this;
};

SocketNioClient.prototype.objPackageFunc = function(obj, isAck) {
  let dataBytes = Buffer.from(JSON.stringify(obj), "utf8");
  return this.packageFunc(dataBytes, isAck);
};

SocketNioClient.prototype.packageFunc = function(data, isAck) {
  let ackBytes = Buffer.alloc(3);
  let nextInt = crypto.randomInt(16777215);
  if (isAck) {
    ackBytes.writeUintBE(0x80 | (nextInt >>> 25), 0, 1);
  } else {
    ackBytes.writeUintBE(0x00 | (nextInt >>> 25), 0, 1);
  }
  ackBytes.writeUintBE(0xff & (nextInt >>> 16), 1, 1);
  ackBytes.writeUintBE(0xff & nextInt, 2, 1);
  // let result = Buffer.concat([ackBytes, data]);
  // console.debug('package:', result)
  // return result;
  return Buffer.concat([ackBytes, data]);
};

SocketNioClient.prototype.secretEncode = function(data) {
  const encryptCipher = crypto.createCipheriv(
    this.secretAlgorithm,
    this.secretBytes,
    this.secretIv
  );
  encryptCipher.setAutoPadding(true);
  return Buffer.concat([encryptCipher.update(data), encryptCipher.final()]);
};

SocketNioClient.prototype.wrapFunc = function(data) {
  let appKeyBytes = this.appKeyBytes;
  let appKeyBytesLength = appKeyBytes.length;
  let msgSize = 4 + appKeyBytesLength + data.length + 8;
  let appKeyLenBytes = Buffer.alloc(4);
  appKeyLenBytes.writeUintBE(appKeyBytesLength, 0, 4);
  let msgSizeBytes = Buffer.alloc(4);
  msgSizeBytes.writeUintBE(msgSize, 0, 4);
  let msgSizeFirstByte = Buffer.of(msgSizeBytes.readUintBE(1, 1));
  let result = Buffer.concat([
    this.headByte,
    msgSizeBytes,
    appKeyLenBytes,
    appKeyBytes,
    data,
    msgSizeFirstByte,
    this.headByte,
    this.secretByte,
  ]);
  result.writeUintBE(result.readUintBE(msgSize / 2, 1), msgSize - 2, 1);
  // console.debug('wrap:', result)
  return result;
};

SocketNioClient.prototype.unPackageFunc = function(packageData) {
  return packageData.slice(3, packageData.length);
};

SocketNioClient.prototype.secretDecode = function(data) {
  let decryptCipher = crypto.createDecipheriv(
    this.secretAlgorithm,
    this.secretBytes,
    this.secretIv
  );
  decryptCipher.setAutoPadding(true);
  return Buffer.concat([decryptCipher.update(data), decryptCipher.final()]);
};

SocketNioClient.prototype.parseFunc = function(cacheMsg, msg) {
  let full = cacheMsg.full;
  let size = cacheMsg.size;
  let readableBytes;
  let writeIndex;
  if (full.length <= 0) {
    readableBytes = msg.length;
    if (readableBytes < 5) {
      //数据不足，继续等待
      cacheMsg.full = Buffer.concat([full, msg]);
      return null;
    }
    //头部，5字节
    let head = msg.slice(0, 5);
    msg = msg.slice(5, msg.length);
    size = cacheMsg.size = this.checkMsgFirst(head);
    //5+2+3+1
    if (size < 11) {
      //丢弃并关闭连接
      throw new Error("非法报文");
    }
    full = cacheMsg.full = Buffer.concat([full, head]);
    writeIndex = full.length;
  } else {
    if (size == null) {
      //补全数据
      let writerIndex = full.length;
      readableBytes = msg.length;
      if (writerIndex < 4) {
        if (writerIndex + readableBytes < 4) {
          //数据不足，继续等待
          cacheMsg.full = Buffer.concat([full, msg]);
          return null;
        }
      }
      //头部，5字节
      let head = msg.slice(0, 5);
      msg = msg.slice(5, msg.length);
      size = cacheMsg.size = this.checkMsgFirst(head);
      if (size < 11) {
        //丢弃并关闭连接
        throw new Error("非法报文");
      }
      full = cacheMsg.full = Buffer.concat([full, head]);
    }
    writeIndex = full.length;
  }
  let stickMsg = null;
  let leftDataLength = size - writeIndex;
  if (leftDataLength > 0) {
    readableBytes = msg.length;
    let i = leftDataLength - readableBytes;
    if (i > 0) {
      //继续读取
      cacheMsg.full = Buffer.concat([full, msg]);
      return null;
    } else if (i < 0) {
      //此处粘包了，手动切割
      full = cacheMsg.full = Buffer.concat([
        full,
        msg.slice(0, leftDataLength),
      ]);
      stickMsg = msg.slice(leftDataLength, msg.length);
    } else {
      full = cacheMsg.full = Buffer.concat([full, msg]);
    }
  } else if (leftDataLength < 0) {
    //丢弃并关闭连接
    throw new Error("报文数据异常");
  }
  cacheMsg.secretByte = this.checkMsgTail(full, size);
  cacheMsg.msg = full.slice(5, size - 3);
  cacheMsg.isDone = true;
  return stickMsg;
};

SocketNioClient.prototype.checkMsgFirst = function(msg) {
  //验证报文头是否为:10100101
  if (this.headByte.compare(Buffer.of(msg.readUintBE(0, 1))) !== 0) {
    return 0;
  }
  let msgSize = msg.readUintBE(1, 4);
  if (msgSize > this.sizeLimit) {
    return 0;
  }
  /*if (log.isDebugEnabled()) {
        log.debug("检查报文 长度:{}，头部:{},{},{},{},{}"
                , msgSize
                , msg.getByte(0)
                , msg.getByte(1)
                , msg.getByte(2)
                , msg.getByte(3)
                , msg.getByte(4)
        );
    }*/
  return msgSize;
};

SocketNioClient.prototype.checkMsgTail = function(full, size) {
  //报文尾共3字节，2字节为 msgSize首字节 + 数据中间字节，1字节加密类型
  if (
    full.readUintBE(size - 3, 1) === full.readUintBE(1, 1) &&
    full.readUintBE(size - 2, 1) === full.readUintBE(size / 2, 1)
  ) {
    return Buffer.of(full.readUintBE(size - 1, 1));
  }
  throw new Error("报文尾部验证失败");
};

SocketNioClient.prototype.isAckData = function(packageData) {
  return 0x01 === (packageData.readUintBE(0, 1) & 0xff) >>> 7;
};

SocketNioClient.prototype.packageAckBytesToInt = function(ackBytes) {
  let needAckBytes = Buffer.alloc(4);
  needAckBytes.writeUintBE(ackBytes.readUintBE(0, 1) & 0x7f, 0, 1);
  needAckBytes.writeUintBE(ackBytes.readUintBE(1, 1), 1, 1);
  let number = ackBytes.readUintBE(2, 1);
  needAckBytes.writeUintBE(number, 2, 1);
  needAckBytes.writeUintBE(number, 3, 1);
  return needAckBytes.readUintBE(0, 4);
};

SocketNioClient.prototype.unPackageAckBytesToInt = function(ackBytes) {
  let needAckBytes = Buffer.alloc(4);
  needAckBytes.writeUintBE(ackBytes.readUintBE(0, 1), 0, 1);
  needAckBytes.writeUintBE(ackBytes.readUintBE(1, 1), 1, 1);
  let number = ackBytes.readUintBE(2, 1);
  needAckBytes.writeUintBE(number, 2, 1);
  needAckBytes.writeUintBE(number, 3, 1);
  return needAckBytes.readUintBE(0, 4);
};

SocketNioClient.prototype.getAckData = function(packageData) {
  let ackBytes = Buffer.alloc(3);
  //把第一位设置成0，即无需接收端再ack
  ackBytes.writeUintBE(packageData.readUintBE(0, 1) & 0x7f, 0, 1);
  ackBytes.writeUintBE(packageData.readUintBE(1, 1), 1, 1);
  ackBytes.writeUintBE(packageData.readUintBE(2, 1), 2, 1);
  return ackBytes;
};

SocketNioClient.prototype.read = function(data, socket) {
  try {
    let leftMsg = data;
    do {
      //接收数据
      let cacheMsg = socket.cacheMsg;
      if (!cacheMsg || cacheMsg.isDone) {
        cacheMsg = {
          msg: null,
          secretByte: null,
          full: Buffer.alloc(0),
          size: null,
          isDone: false,
          isRelease: false,
        };
        socket.cacheMsg = cacheMsg;
      }
      leftMsg = this.parseFunc(cacheMsg, leftMsg);
      if (!cacheMsg.isDone) {
        return;
      }
      //读取完成，写入队列
      let decodeBytes;
      try {
        decodeBytes = this.secretDecode(cacheMsg.msg);
      } catch (e) {
        console.error("解码错误", e.message);
        //丢弃并关闭连接
        throw new Error("报文解码错误");
      }
      let length = decodeBytes.length;
      let sendOrReceiveAck = this.isAckData(decodeBytes);
      if (length > 3) {
        console.debug("数据已接收");
        let result = this.unPackageFunc(decodeBytes).toString("utf8");
        runner.add(async (done) => {
          let time = new Date().getTime();
          taskMap[time] = done;

          await this.consumer(result)
            .catch((reason) => {
              console.error("数据处理错误", reason.message);
            })
            .finally(() => {
              delete taskMap[time];
              done();
            });
        });
        runner.start()
        if (sendOrReceiveAck) {
          let ackBytes = this.getAckData(decodeBytes);
          try {
            this.writeBytes(ackBytes);
          } catch (e) {
            console.error("ack编码错误", e);
          }
        }
      } else {
        if (!sendOrReceiveAck) {
          //接收ackBytes
          let ackBytes = this.getAckData(decodeBytes);
          let ackKey = this.unPackageAckBytesToInt(ackBytes);
          let ackPromiseCallback = this.ackPromiseCallbackMap[ackKey];
          if (ackPromiseCallback) {
            console.log("已接收ack字节");
            ackPromiseCallback(true);
          } else {
            console.error(
              "接收ack字节:",
              JSON.stringify(ackBytes),
              "未命中或请求超时"
            );
          }
        } else {
          //关闭连接
          throw new Error("报文数据异常");
        }
      }
      socket.cacheMsg = null;
      cacheMsg = null;
    } while (leftMsg != null);
  } catch (e) {
    console.error("数据解析异常:", e.message);
    socket.cacheMsg = null;
    socket.destroy(e);
  }
};

SocketNioClient.prototype.consumer = async function(bytes) {
  let resSocketData = JSON.parse(bytes);
  let clientDataId = resSocketData.clientDataId;
  if (clientDataId != null) {
    let resolve = this.syncPromiseCallbackMap[clientDataId];
    if (resolve) {
      if (resSocketData.code === 200) {
        resolve(resSocketData.nodeData.node);
      } else {
        console.error("服务端响应异常:", JSON.stringify(resSocketData));
        resolve();
      }
    }
    //不再继续执行，即便method不为空
    return;
  }
  let method = resSocketData.method;
  if (method && method !== "") {
    if (resSocketData.code === 200) {
      let serverDataId = resSocketData.serverDataId;
      let syncData;
      try {
        let data;
        console.debug("客户端执行方法:", method);
        let methodFunc = this.methodMap[method];
        if (methodFunc) {
          if (
            Object.prototype.toString.call(methodFunc) ===
              "[object AsyncFunction]" ||
            methodFunc
              .toString()
              .includes("return _regenerator.default.async(function")
          ) {
            data = await methodFunc(resSocketData.nodeData.node);
          } else {
            data = methodFunc(resSocketData.nodeData.node);
          }
          syncData = {
            code: 200,
            method: null,
            clientDataId: null,
            serverDataId: null,
            nodeData: {
              node: data,
            },
          };
        } else {
          syncData = {
            code: 500,
            msg: "方法未注册",
            method: null,
            clientDataId: null,
            serverDataId: null,
          };
        }
      } catch (e) {
        syncData = {
          code: 500,
          msg: e.message,
          method: null,
          clientDataId: null,
          serverDataId: null,
        };
      }
      if (serverDataId != null) {
        syncData.serverDataId = serverDataId;
        let bytes = this.objPackageFunc(syncData, false);
        this.writeBytes(bytes);
      }
    } else {
      console.error("服务端响应异常:", JSON.stringify(resSocketData));
    }
  }
};

SocketNioClient.prototype.writeBytes = function(bytes) {
  let encode = this.wrapFunc(this.secretEncode(bytes));
  this.pool.pull(function(err, conn) {
    if (err) {
      console.error("获取连接失败，数据发送失败", err.message);
      throw err;
    }
    //发送数据
    conn.write(encode);
    console.log("数据发送成功");
  });
};

SocketNioClient.prototype.writeSync = async function(
  method,
  data,
  timeout = 30000
) {
  let dataId = crypto.randomBytes(4).readIntBE(0, 4);
  let socketData = {
    code: 200,
    method: method,
    clientDataId: dataId,
    serverDataId: null,
    nodeData: {
      node: data,
    },
  };
  let bytes = this.objPackageFunc(socketData, false);
  let setTimeoutId = null;
  try {
    return await new Promise((resolve, reject) => {
      this.writeBytes(bytes);
      this.syncPromiseCallbackMap[dataId] = resolve;
      setTimeoutId = setTimeout(() => resolve(), timeout);
    });
  } finally {
    clearTimeout(setTimeoutId);
    delete this.syncPromiseCallbackMap[dataId];
  }
};

SocketNioClient.prototype.writeAck = async function(
  method,
  data,
  timeout = 30000
) {
  let socketData = {
    code: 200,
    method: method,
    clientDataId: crypto.randomBytes(4).readIntBE(0, 4),
    serverDataId: null,
    nodeData: {
      node: data,
    },
  };
  let bytes = this.objPackageFunc(socketData, true);
  let ackKey = this.packageAckBytesToInt(bytes);
  let setTimeoutId = null;
  try {
    return await new Promise((resolve, reject) => {
      this.writeBytes(bytes);
      this.ackPromiseCallbackMap[ackKey] = resolve;
      setTimeoutId = setTimeout(() => resolve(false), timeout);
    });
  } finally {
    clearTimeout(setTimeoutId);
    delete this.ackPromiseCallbackMap[ackKey];
  }
};

SocketNioClient.prototype.write = function(method, data) {
  let socketData = {
    code: 200,
    method: method,
    clientDataId: crypto.randomBytes(4).readIntBE(0, 4),
    serverDataId: null,
    nodeData: {
      node: data,
    },
  };
  let bytes = this.objPackageFunc(socketData, false);
  this.writeBytes(bytes);
};

SocketNioClient.prototype.addMethod = function(method, methodFunc) {
  this.methodMap[method] = methodFunc;
};

SocketNioClient.prototype.close = function() {
  if (this.pool) {
    clearInterval(this.intervalId);
    this.pool.end(true);
  }
};

/*SocketNioClient.prototype.on = function (event, listener) {
    this.socket.on(event, listener);
}*/

module.exports = () => {
  //初始化SocketNio客户端
  return new SocketNioClient();
};
